package com.clj.rocketmq_example.mq.listener;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.common.message.MessageExt;

import com.clj.rocketmq_example.annotation.RocketMqListener;
import com.clj.rocketmq_example.enums.MqAction;
import com.clj.rocketmq_example.mq.MessageListener;
import com.clj.rocketmq_example.mq.command.MqConstant;
import com.clj.rocketmq_example.util.MqMsgConvertUtil;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@RocketMqListener(topic = "TopicTest", consumerGroup = MqConstant.STAFF_TOPIC_GROUP)
public class ResultListener implements MessageListener{
	
	@Override
	public MqAction consume(MessageExt message, ConsumeConcurrentlyContext context) {
		 String msg = null;
	     System.out.println("===========我是消费者1111111===========" + message);
	     try {
	            msg = MqMsgConvertUtil.bytes2String(message.getBody(), "UTF-8");

	            log.info("MsgId:{},MQ消费,Topic:{},Tag:{}，Body:{}", message.getMsgId(), message.getTopic(), message.getTags(), msg);

	        } catch (Exception e) {
	            log.error("MsgId:{},应用MQ消费失败,Topic:{},Tag:{}，Body:{},异常信息:{}", message.getMsgId(), message.getTopic(), message.getTags(), msg, e);
	            throw e;
	        }
	        return MqAction.CommitMessage;
	}

}
